#include <amqpcpp.h>
#include <amqpcpp/linux_tcp.h>  
#include <amqpcpp/libev.h>      
#include <memory>               
#include <string>               
#include <iostream>
#include "Logger.hpp"
#include <functional>
#include <thread>

namespace chatroom{
    // 封装RabbitMq接口
    class MqClient{
        public:
            using Ptr = std::shared_ptr<MqClient>;
            using MsgCallback = std::function<void(const char*,size_t)>;
            MqClient(const std::string& user,const std::string& pswd,const std::string& host)
            {
                _loop = EV_DEFAULT;
                _hander = std::make_unique<AMQP::LibEvHandler>(_loop);
                //AMQP::Address("amqp://root:123456@127.0.0.1:5672/"));
                AMQP::Address addr = "amqp://"+user+":" + pswd + "@" + host + "/";
                _connection = std::make_unique<AMQP::TcpConnection>(_hander.get(),addr);
                _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());
                bool ret = _channel->connected();
                if(ret==false){
                    std::cout<<"创建信道失败"<<std::endl;
                }

                _loop_thread = std::thread([this](){
                    ev_run(_loop);
                });

            }
            //指定交换机 队列 进行绑定
            void EQbind(const std::string& exchange,const std::string& queue,AMQP::ExchangeType exchangetype = AMQP::ExchangeType::direct,const std::string& routingkey="routing-key")
            {
                // 指定交换机类型
                _channel->declareExchange(exchange,exchangetype)
                    .onError([exchange](const char* message){
                    Logger_ERROR("{}交换机创建失败,{}",exchange,message);
                    })
                    .onSuccess([exchange](){
                        Logger_INFO("{}交换机创建成功",exchange);
                    });
                // 指定队列类型
                _channel->declareQueue(queue)
                .onError([queue](const char* message){
                    Logger_ERROR("{}队列创建失败,{}",queue,message);
                    })
                    .onSuccess([queue](){
                        Logger_INFO("{}队列创建成功",queue);
                    });
                // 绑定交换机和队列
                _channel->bindQueue(exchange,queue,routingkey)
                .onError([exchange,queue](const char* message){
                    Logger_ERROR("{}-{}绑定失败,{}",exchange,queue,message);})
                    .onSuccess([exchange,queue](){
                    Logger_INFO("{}-{}绑定成功",exchange,queue);});
            }
            ~MqClient(){
                struct ev_async async_watcher;
                ev_async_init(&async_watcher,CALLBACK);
                ev_async_start(_loop,&async_watcher);
                ev_async_send(_loop,&async_watcher);
                _loop_thread.join();
            }
            void Publish(const std::string& exchange,const std::string& message,const std::string& routingkey="routing-key")
            {   
                bool ret = _channel->publish(exchange,routingkey,message);
                if(ret==false)
                {
                Logger_ERROR("消息发送失败");
                return;
                }
                Logger_INFO("消息发送成功:{}",message);

            }

            void Consume(const std::string& queue,const MsgCallback& cb)
            {
                _channel->consume(queue,"consume-tag").onReceived([this,cb](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered){
                    cb(message.body(),message.bodySize());
                    _channel->ack(deliveryTag,0);
                })
                .onError([](const char* message){
                Logger_ERROR("订阅消息失败{}",message);
                });
            }
        private:
            static void CALLBACK(struct ev_loop *loop, ev_async *watcher, int32_t revents)
            {
                ev_break(loop,revents);
            }

        private:
            struct ev_loop * _loop;
            std::unique_ptr<AMQP::LibEvHandler> _hander;
            std::unique_ptr<AMQP::TcpConnection> _connection;
            std::unique_ptr<AMQP::TcpChannel> _channel;
            std::thread _loop_thread;
    };
}